/**
 * 
 */
package database.berkeleydb.perf;

import com.sleepycat.je.CursorConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.Transaction;
import com.sleepycat.persist.EntityCursor;
import com.sleepycat.persist.EntityStore;
import com.sleepycat.persist.PrimaryIndex;

/**
 * @author LionBule
 * 2011-5-10 ����02:59:29
 */
public class MyThread extends Thread {
	 /* Global settings */
    protected static final int EXIT_SUCCESS = 0;
    protected static final int EXIT_FAILURE = 1;

    /* Thread id */
    protected int id;
    /* Start time */
    private long start;
    /* End time */
    private long end;
    /* Num ops required */
    private int numRequiredOps;
    /* Num txns completed */
    private int numTxns;
    /* Num operations executed */
    private int numExecutedOps;
    /* Num deadlocks seen, results in retries */
    private int numDeadlocks;
    /* Entitystore */
    private EntityStore store;
    /* Primary index */
    private PrimaryIndex primaryIndex;
    /* Runtime configurations */
    private MyExample.TestConfig testConfig;

    private int keyNum;

    public MyThread(int id,
                    EntityStore store,
                    PrimaryIndex primaryIndex,
                    MyExample.TestConfig testConfig,
                    int numRequiredOps) {
        this.id = id;
        this.store = store;
        this.primaryIndex = primaryIndex;
        this.testConfig = testConfig;
        this.numRequiredOps = numRequiredOps;
    }

    public void run() {

        System.out.println("Access method thread: " + id + " started.");

        /* Allow the other threads to start. */
        yield();

        /* Record our start time. */
        start = System.currentTimeMillis();

        numTxns = 0;
        int orderedKey = 0;

        try {
            numExecutedOps = 0;
            boolean done = false;

            orderedKey = id;
            while (!done) {
                Object txn = null;
                try {
                    /* Begin the transaction. */
                    if (true) {
                        txn = beginTransaction();
                        numTxns++;
                    }

                    /* Perform the operations. */
                    boolean execOk = true;
                    int itemsPerTxn = testConfig.itemsPerTxn;
                    for (int j = 0; j < itemsPerTxn; j++) {
                        selectKey(orderedKey);

                        if (testConfig.operationType.equalsIgnoreCase("READ")) {
                            /* Read */
                            databaseGet(txn);
                        } else if (testConfig.operationType.
                                equalsIgnoreCase("DELETE")) {
                            /* Delete */
                            databaseDelete(txn);
                        } else if (testConfig.operationType.
                                equalsIgnoreCase("INSERT")) {
                            /* Insert */
                            databasePutNoOverwrite(txn);
                        } else if (testConfig.operationType.
                                equalsIgnoreCase("UPDATE")) {
                            /* Update */
                            databaseUpdate(txn);
                        } else {
                            /* Cursor scan with dirty read. */
                            scan();
                        }

                        if (execOk) {
                            numExecutedOps++;
                            orderedKey += testConfig.numThreads;
                        }

                        /* See if we're done. */
                        if (numExecutedOps >= numRequiredOps) {
                            done = true;
                        }

                        if (done) {
                            break;
                        }
                    }

                    /* Commit the transaction. */
                    if (txn != null) {
                        commitTxn(txn);
                    }

                    if (done) {
                        break;
                    }
                } catch (Exception e) {
                    /* Deal with deadlock and other errors. */
                    if (txn != null) {
                        try {
                            abortTxn(txn);
                        } catch (Exception e2) {
                            System.err.println("abort: " + e2);
                            System.err.println("original exception: " + e);
                            System.exit(EXIT_FAILURE);
                        }
                    }
                    exitIfNotDeadlock(e);
                    /* else will retry on next iteration */
                    numDeadlocks++;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }

        /* Record our end time. */
        end = System.currentTimeMillis();

        System.out.println("Access method thread " + id +
                " exiting cleanly. Executed " + numRequiredOps + " " +
                testConfig.operationType + " ops in " +
                (end - start) + " ms.");
    }

    /**
     * Select the key. It may be chosen in one of four ways:
     *  - random value
     *  - value within a range
     *  - value tied to the current iteration
     *  - a single, present value.
     *
     */
    private void selectKey(int iterationVal) {
        int val = iterationVal;

        assignKey(val);
    }

    private void assignKey(int val) {
        keyNum = val;
    }

    private Object beginTransaction()
        throws Exception {

        try {
            Transaction txn =
                store.getEnvironment().beginTransaction(null, null);
            return txn;
        } catch (DatabaseException DE) {
            System.out.println("Caught " + DE + " during beginTransaction");
            return null;
        }
    }

    private void abortTxn(Object txn)
        throws Exception {

        Transaction t = (Transaction) txn;

        try {
            t.abort();
        } catch (DatabaseException DE) {
            System.out.println("Caught " + DE + " during abort");
        }
    }

    private void commitTxn(Object txn)
        throws Exception {

        Transaction t = (Transaction) txn;

        try {
            if (testConfig.syncCommit) {
                t.commitSync();
            } else {
                t.commitNoSync();
            }
        } catch (DatabaseException DE) {
            System.out.println("Caught " + DE + " during abort");
        }
    }

    private void exitIfNotDeadlock(Exception e) {

        System.err.println("unexpected exception: " + e);
        e.printStackTrace();
        System.exit(EXIT_FAILURE);
    }

    private void databasePutNoOverwrite(Object txn)
        throws Exception {

        BasicEntity entity = new BasicEntity(keyNum);
        primaryIndex.putNoReturn((Transaction) txn, entity);
    }

    private void databaseGet(Object txn) throws Exception  {

        BasicEntity e = (BasicEntity)primaryIndex.get((Transaction) txn, new CompositeKey(keyNum), LockMode.DEFAULT);
        // System.out.println("Read entity = " + e);
    }

    private void databaseUpdate(Object txn) throws Exception {

        Transaction t = (Transaction) txn;
        BasicEntity entity = (BasicEntity)primaryIndex.get(t, new CompositeKey(keyNum), LockMode.DEFAULT);
        entity.modify();
        primaryIndex.putNoReturn((Transaction) txn, entity);
    }

    private int scan() throws Exception {

        int numRecords = 0;
        CursorConfig curConf = new CursorConfig();
        if (true) {
            curConf.setReadUncommitted(true);
            System.out.println("setting dirty read");
        }

        Object txn = null;
        if (true) {
            txn = beginTransaction();
        }

        EntityCursor cursor =
            primaryIndex.entities((Transaction) txn, curConf);
        try {
            for (Object e : cursor) {
                numRecords++;
            }
        } finally {
            cursor.close();
        }

        if (true) {
            commitTxn(txn);
        }

        System.out.println("scan records=" + numRecords);
        return numRecords;
    }

    private void databaseDelete(Object txn)
        throws Exception {
        primaryIndex.delete((Transaction) txn, new CompositeKey(keyNum));
    }	
}
